package com.atguigu.day04;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import com.atguigu.util.UserViewCountPerWindow;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;

public class Example7 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.username)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .aggregate(
                        new AggregateFunction<ClickEvent, Integer, Integer>() {
                            @Override
                            public Integer createAccumulator() {
                                return 0;
                            }

                            @Override
                            public Integer add(ClickEvent value, Integer accumulator) {
                                return accumulator + 1;
                            }

                            @Override
                            public Integer getResult(Integer accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Integer merge(Integer a, Integer b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Integer, UserViewCountPerWindow, String, TimeWindow>() {
                            @Override
                            public void process(String key, Context context, Iterable<Integer> iterable, Collector<UserViewCountPerWindow> collector) throws Exception {
                                collector.collect(new UserViewCountPerWindow(
                                        key,
                                        iterable.iterator().next(),
                                        context.window().getStart(),
                                        context.window().getEnd()
                                ));
                            }
                        }
                )
                // 再进行一次分组
                .keyBy(new KeySelector<UserViewCountPerWindow, Tuple2<Long, Long>>() {
                    @Override
                    public Tuple2<Long, Long> getKey(UserViewCountPerWindow value) throws Exception {
                        return Tuple2.of(value.windowStartTime, value.windowEndTime);
                    }
                })
                .process(new KeyedProcessFunction<Tuple2<Long, Long>, UserViewCountPerWindow, String>() {
                    private ListState<UserViewCountPerWindow> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        listState = getRuntimeContext().getListState(
                                new ListStateDescriptor<UserViewCountPerWindow>(
                                        "userViewCountPerWindow-list",
                                        Types.POJO(UserViewCountPerWindow.class)
                                )
                        );
                    }

                    // 每来一个UserViewCountPerWindow,就降序排列输出第一名
                    @Override
                    public void processElement(UserViewCountPerWindow in, Context context, Collector<String> collector) throws Exception {
                        listState.add(in);

                        var arrayList = new ArrayList<UserViewCountPerWindow>();

                        for (var e : listState.get()) {
                            arrayList.add(e);
                        }

                        arrayList.sort(new Comparator<UserViewCountPerWindow>() {
                            @Override
                            public int compare(UserViewCountPerWindow o1, UserViewCountPerWindow o2) {
                                return o2.count - o1.count;
                            }
                        });

                        var windowInfo = context.getCurrentKey();
                        collector.collect(new Timestamp(windowInfo.f0) + "~" + new Timestamp(windowInfo.f1) + "浏览次数最多的用户是：" +
                                "" + arrayList.get(0).username + ", 浏览次数是：" + arrayList.get(0).count);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
